package com.resoft.bigdata.p2p.job;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.resoft.bigdata.p2p.record.MonthlyStatistics;
import com.resoft.bigdata.p2p.service.api.QueryService;
import com.resoft.bigdata.p2p.service.api.thrift2imp.Thrift2QueryServiceImpl;

public class DailyReportStatisticReducer {
	private static final Logger logger = LoggerFactory.getLogger(DailyReportStatisticReducer.class);
	
	private List<StatisticJobListenner> listeners = new ArrayList<StatisticJobListenner>();
	
	private BlockingQueue<String> orgCodes = new ArrayBlockingQueue<String>(1000);
//	private ConcurrentHashMap<String, MonthlyStatistics> staMap = new ConcurrentHashMap<String, MonthlyStatistics>();
	private List<MonthlyStatistics> monthlyStatistics = new ArrayList<MonthlyStatistics>();
	
	private QueryService qry = null;
	private String host;
	private int port;
	private int threadNumber = 2;
	private String beginDate;
	private String endDate;
	private AtomicInteger remainderThreads;
	private int yearAndMonth = 0;
	
	public DailyReportStatisticReducer(String host, int port, Date end){
		this.host = host;
		this.port = port;
		this.threadNumber = Runtime.getRuntime().availableProcessors()*2;
		remainderThreads = new AtomicInteger(threadNumber);
		System.out.println("============  =========== Thread number:" + threadNumber);
		qry =  new Thrift2QueryServiceImpl(host, port);
		
//		Date begin = new Date(end.getTime()-30*24*3600*1000);
		
		//caculate beginDate，endDate
		Calendar cal=Calendar.getInstance();
		cal.setTime(end);
		int year = cal.get(Calendar.YEAR);
		int month = cal.get(Calendar.MONTH)+1;
		int day = cal.get(Calendar.DATE);
		yearAndMonth = year*10000+month*100+day;
		
		cal.add(Calendar.DATE, -30);
		Date begin = cal.getTime();
		
		SimpleDateFormat format=new SimpleDateFormat("yyyy-MM-dd");  
		
		beginDate = format.format(begin);
		endDate = format.format(end);
		
	}
	
	public void run(){
		for(StatisticJobListenner listener: listeners){
			listener.beginToStatistic(beginDate, endDate);
		}
		
		new QueryOrgCodes().run();
		
		if(orgCodes.size()<threadNumber) threadNumber = orgCodes.size();
		
		Thread[] threads = new Thread[threadNumber];
		for(int i=0;i<threadNumber;i++ ){
			threads[i] = new Thread(new CalulateMonthlyStatistic());
			threads[i].start();
		}
		
		try {
			for(int i=0;i<threadNumber;i++){
				threads[i].join();
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		
		for(StatisticJobListenner listener: listeners){
			listener.finishStatistic(beginDate, endDate, monthlyStatistics);
		}
		
	}
	
	public void addListener(StatisticJobListenner listener){
		listeners.add(listener);
	}
	
	public void removeListener(StatisticJobListenner listener){
		listeners.remove(listener);
	}
	
	public void removeAllListeners(){
		listeners.clear();
	}
	
	class QueryOrgCodes implements Runnable{
		@Override
		public void run() {
			try {
				List<String> p2pOrgCodes = qry.getOrgCodes();
				System.out.println("ALl p2p size:" + p2pOrgCodes.size());
				for(String orgCode: p2pOrgCodes){
					orgCodes.put(orgCode);
				}
				for(int i=0; i<threadNumber; i++){
					orgCodes.put("EOF");
				}
			} catch (IOException e) {
				logger.error(e.getMessage());
				e.printStackTrace();
			} catch (InterruptedException e) {
				logger.error(e.getMessage());
				e.printStackTrace();
			}
			

		} // ~~ run
		
	} //~~ QueryOrgCodes
	
	class CalulateMonthlyStatistic implements Runnable{

		@Override
		public void run() {
			DailyReportStatistic sta = new DailyReportStatistic(host,  port);
			while(true){
				String orgCode = null;
				try {
					orgCode = orgCodes.take();
					if(orgCode.equalsIgnoreCase("EOF")){
						logger.info("CalulateMonthlyStatistic finished");
						break;
					}
					System.out.println("Thread[" + Thread.currentThread().getId() + "] calculate : " + orgCode);
					logger.debug("Thread[{}] Begin to calculate MonthlyStatistic, p2p orgCode{}", Thread.currentThread().getId(), orgCode);
					MonthlyStatistics ms = sta.getMonthlyStatistics(orgCode, beginDate, endDate);
					ms.setYearAndMonth(yearAndMonth);
					monthlyStatistics.add(ms);
					logger.debug("Thread[{}] Finished to calculate MonthlyStatistic, p2p orgCode{}", Thread.currentThread().getId(), orgCode,ms.getOrgCode());

				}catch (InterruptedException e) {
					System.out.println(">>>>>> Thread[" + Thread.currentThread().getId() + "] exception:" + e.getMessage()); 
					logger.error("CalulateMonthlyStatistic failed, Error message{}", e.getMessage());
					e.printStackTrace();
					for(StatisticJobListenner listener: listeners){
						listener.failedStatistic(orgCode, beginDate, endDate,  e.getMessage());
					}
				} 
			} // ~~while
			
			remainderThreads.addAndGet(-1);
			System.out.println("---------------- Thread[" + Thread.currentThread().getId() + "] exit!!!, remaind:" + remainderThreads.get());
		}
		
	} // ~~ class
	
}
